package com.bw.ads;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.VisitorstatsEntity;
import com.bw.utils.ClickHouseUtil;
import com.bw.utils.DateFormatUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;

/*
求每10秒网站 pv,uv,jv ,并写入clickhouse
Create table visitorstats (stt String,edt String,is_new String ,pv Int64,uv Int64,jv Int64,ts Int64) engine=MergeTree() order by stt;
 */
public class XT4Dws {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092");
        properties.setProperty("group.id", "group1");
        DataStream<String> stream1 = env.addSource(MyKafkaUtil.getKafkaConsumer("dwd_page_log_yk2","test115"));
        DataStream<String> stream2 = env.addSource(MyKafkaUtil.getKafkaConsumer("dwm_user_jump_yk2","test115"));
        stream2.print("跳出明细：");
        //PV {common:{} page:{}  ts:}
        SingleOutputStreamOperator<VisitorstatsEntity> pvDS = stream1.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                JSONObject jo = JSON.parseObject(s);
                return jo.getJSONObject("common").getString("mid");
            }
        }).map(new RichMapFunction<String, VisitorstatsEntity>() {
            private ValueState<String> state;
            @Override
            public void open(Configuration parameters) throws Exception {
                state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", String.class));
            }

            @Override
            public VisitorstatsEntity map(String s) throws Exception {
                String value = state.value();
                JSONObject jo = JSON.parseObject(s);
                Long ts = jo.getLong("ts");
                if (value == null){
                    state.update(DateFormatUtil.tsToDate(ts));
                    return new VisitorstatsEntity(
                            null, null, jo.getJSONObject("common").getString("is_new"),1L, 1L, 0L, jo.getLong("ts")
                    );
                }else if(!DateFormatUtil.tsToDate(ts).equals(value)){
                    state.update(DateFormatUtil.tsToDate(ts));
                    return new VisitorstatsEntity(
                            null, null, jo.getJSONObject("common").getString("is_new"),1L, 1L, 0L, jo.getLong("ts")
                    );
                }else{
                    return new VisitorstatsEntity(
                            null, null, jo.getJSONObject("common").getString("is_new"),1L, 0L, 0L, jo.getLong("ts")
                    );
                }


            }
        });
        //JV
        SingleOutputStreamOperator<VisitorstatsEntity> jvDS = stream2.map(new MapFunction<String, VisitorstatsEntity>() {
            @Override
            public VisitorstatsEntity map(String s) throws Exception {
                JSONObject jo = JSON.parseObject(s);
                return new VisitorstatsEntity(
                        null, null, jo.getJSONObject("common").getString("is_new"), 0L, 0L, 1L, jo.getLong("ts")
                );
            }
        });
        DataStream<VisitorstatsEntity> unionDS = pvDS.union(jvDS);


        //每10统计
        SingleOutputStreamOperator<VisitorstatsEntity> resultDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy
                        .<VisitorstatsEntity>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((event, timestamp) -> event.getTs()))
                .keyBy(x -> x.getIs_new())
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new ProcessWindowFunction<VisitorstatsEntity, VisitorstatsEntity, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<VisitorstatsEntity, VisitorstatsEntity, String, TimeWindow>.Context context, Iterable<VisitorstatsEntity> iterable, Collector<VisitorstatsEntity> collector) throws Exception {
                        VisitorstatsEntity result = new VisitorstatsEntity(null, null,null, 0L, 0L, 0L, null);
                        for (VisitorstatsEntity i : iterable) {
                            result.setIs_new(i.getIs_new());
                            result.setPv(result.getPv() + i.getPv());
                            result.setUv(result.getUv() + i.getUv());
                            result.setJv(result.getJv() + i.getJv());
                            result.setTs(i.getTs());
                        }
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        result.setStt(sdf.format(context.window().getStart()));
                        result.setEdt(sdf.format(context.window().getEnd()));
                        collector.collect(result);
                    }
                });

        resultDS.print();
        resultDS.addSink(ClickHouseUtil.sink("insert into view_yk2 (stt,edt,is_new,pv,uv,jv,ts) values(?,?,?,?,?,?,?)"));
        env.execute();



        //select is_new, sum(pv),sum(uv),sum(jv),sum(jv)/sum(pv) from view_yk8 where substring(stt,1,10) = '2020-12-18' group by is_new;

//        select source sum(cnt) cnts from kw group by source
    }
}
